package cn.itcast.flink.exactlyonce;

import lombok.SneakyThrows;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Arrays;

/**
 * Author itcast
 * Date 2022/1/17 11:28
 * Desc 将读取 Socket 的数据源，对其进行wordcount统计之后，将数据通过 TwoPhaseCommitFunction 方式提交到 MySQL数据库中
 * 实现四个方法：
 * beginTransaction 开启事务
 * precommit 预提交
 * commit 提交
 * abort  丢弃，删除失败的数据
 */
public class FlinkMySQLWriter {
    public static void main(String[] args) throws Exception {

        System.setProperty("HADOOP_USER_NAME", "root");
        //todo 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置当前全局的并行度
        env.setParallelism(1);
        //todo 设置chk 1s 状态后端到hdfs或本地file，并行度
        //开启 checkpoint ，按照 1s 将全局的状态保存到 checkpoint 中
        env.enableCheckpointing(1000);
        //先获取读写 checkpoint 配置
        CheckpointConfig conf = env.getCheckpointConfig();
        //设置 hdfs ，将 全局的checkpoint保存到 hdfs 上
        conf.setCheckpointStorage("hdfs://node1:8020/flink-checkpoints/");
        //todo 设置chk属性配置，仅一次模式、超时、并行、容忍、最小间隔、取消任务保存chk
        //设置 checkpoint 的超时时间，超过1分钟
        conf.setCheckpointTimeout(60 * 1000);
        //设置当前checkpoint的执行模式是 仅一次语义
        conf.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //设置checkpoint并行数为 1
        conf.setMaxConcurrentCheckpoints(1);
        //设置容忍失败的checkpoint的次数，如果超过这个次数，程序job会跟着checkpoint失败
        conf.setTolerableCheckpointFailureNumber(10);
        //设置两个 checkpoint 之间的最短间隔时间
        conf.setMinPauseBetweenCheckpoints(500);
        //设置 checkpoint 在job取消的时候，是否删除checkpoint，在HDFS上存储 /flink-checkpoints/ metastore 元数据
        conf.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //todo 设置重启策略 3次，10s间隔
        //设置重启策略，固定重启策略，重启 3次， 每次之间间隔 5秒
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));

        //todo 开启socket数据源
        DataStreamSource<String> source = env.socketTextStream("node1", 9999);
        //todo wordcount求和生成 Tuple2<String,Integer>
        SingleOutputStreamOperator<Tuple2<String, Integer>> result = source.flatMap((String line, Collector<Tuple2<String, Integer>> ctx) -> {
            Arrays.stream(line.split(" ")).map(x -> Tuple2.of(x, 1)).forEach(ctx::collect);
        }).returns(Types.TUPLE(Types.STRING, Types.INT))
                // 5）对每个单词进行分组聚合操作
                .keyBy(t -> t.f0)
                .sum(1);

        //todo 将数据实时写入到MySQL通过二段提交方式
        result.addSink(new MySink());
        //todo 执行流环境
        env.execute();

    }

    //todo 继承TwoPhaseCommitSinkFunction<Tuple2<String, Integer>, ConnectionState, Void>
    public static class MySink extends TwoPhaseCommitSinkFunction<Tuple2<String, Integer>, ConnectionState, Void> {
        public MySink() {
            //todo 构造方法 super(new KryoSerializer<>(ConnectionState.class, new ExecutionConfig()), VoidSerializer.INSTANCE);
            super(
                    new KryoSerializer<>(ConnectionState.class, new ExecutionConfig()),
                    VoidSerializer.INSTANCE
            );
        }

        //todo 重写invoke方法
        @Override
        protected void invoke(ConnectionState transaction, Tuple2<String, Integer> value, Context context) throws Exception {
            //通过state获取连接，通过连接获取预编译状态SQL
            Connection conn = transaction.conn;
            //编写插入到 t_wordcount sql
            String sql = "INSERT INTO t_wordcount (word, counts) VALUES (?, ?) on duplicate key update counts=?";
            //获取 preparedStatement 预编译对象
            PreparedStatement ps = conn.prepareStatement(sql);
            //"INSERT INTO t_wordcount (word, counts) VALUES (?, ?) on"
            //设置每个参数
            ps.setString(1, value.f0);
            ps.setInt(2, value.f1);
            ps.setInt(3, value.f1);
            //执行更新并关闭
            ps.executeUpdate();
            //手动制造异常，如果当前f0=hive 抛个 1/0 错误。
            if (value.f0.equalsIgnoreCase("hive")) {
                throw new RuntimeException("1/0");
            }
        }

        //todo 重写 beginTransaction 方法
        @Override
        protected ConnectionState beginTransaction() throws Exception {
            //设置驱动类，创建连接，设置连接不自动提交，返回连接状态
            Class.forName("com.mysql.jdbc.Driver");
            Connection conn = DriverManager.getConnection(
                    "jdbc:mysql://node3:3306/flink?useSSL=false",
                    "root",
                    "123456"
            );
            //设置事务机制不要自动提交给数据库，将事务自动提交设置为 false
            conn.setAutoCommit(false);
            //将 conn 连接器给 ConnectionState
            ConnectionState connectionState = new ConnectionState(conn);
            return connectionState;
        }

        //todo 重写 preCommit 方法 给个提示
        @Override
        protected void preCommit(ConnectionState transaction) throws Exception {
            System.out.println("预编译提交到数据库中");
            Connection conn = transaction.conn;
            System.out.println("conn state" + conn);
        }

        //todo 重写 commit 方法
        @SneakyThrows
        @Override
        protected void commit(ConnectionState transaction) {
            //获取连接，执行提交和连接关闭
            Connection conn = transaction.conn;
            //执行提交
            conn.commit();
            //执行完毕关闭连接
            if (!conn.isClosed()) {
                conn.close();
            }
        }

        //todo 重写 abort 方法
        @SneakyThrows
        @Override
        protected void abort(ConnectionState transaction) {
            //获取连接，回滚并关闭连接
            Connection conn = transaction.conn;
            //失败了，回滚数据
            conn.rollback();
            //关闭连接
            if (!conn.isClosed()) {
                conn.close();
            }
        }
    }

    //todo 静态内部类 ConnectionState
    static class ConnectionState {
        //创建变量 Connection,
        Connection conn = null;

        //构造方法传参
        public ConnectionState(Connection _conn) {
            this.conn = _conn;
        }
    }

}
